package fun.tan90.easy.binlog.plugin.kafka.config;

import com.github.benmanes.caffeine.cache.CacheLoader;
import fun.tan90.easy.binlog.plugin.common.constants.DamiEventConstants;
import fun.tan90.easy.binlog.plugin.common.model.OutputParam;
import fun.tan90.easy.binlog.plugin.kafka.model.KafkaClientParams;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.checkerframework.checker.nullness.qual.NonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.noear.dami.Dami;

import java.util.Objects;
import java.util.Properties;

/**
 * description 动态连接mqtt
 * company 铁人科技
 *
 * @author Li YuLong
 * @date 2023/8/17 10:56
 */
@Slf4j
public class KafkaClientCacheLoader implements CacheLoader<String, KafkaProducer<String, String>> {

    @Override
    public @Nullable KafkaProducer<String, String> load(@NonNull String outputCode) throws Exception {
        OutputParam outputParam = Dami.<String, OutputParam>bus().sendAndRequest(DamiEventConstants.OUTPUT_PARAM_TOPIC, outputCode);
        if (Objects.isNull(outputParam)) {
            return null;
        }
        KafkaClientParams kafkaClientParams = (KafkaClientParams) outputParam;
        Properties props = new Properties();
        /**
         * bootstrap.servers（必需）：指定 Kafka 集群的地址列表，格式为 "host1:port1,host2:port2,..."。生产者将使用这个参数来连接到 Kafka 集群。例如：bootstrap.servers=localhost:9092。
         * key.serializer 和 value.serializer（必需）：指定键（key）和值（value）的序列化器类。生产者需要将消息中的键和值序列化为字节流以进行传输。这两个参数的值通常设置为实现了 org.apache.kafka.common.serialization.Serializer 接口的序列化器类。例如：key.serializer=org.apache.kafka.common.serialization.StringSerializer。
         * acks：指定生产者等待服务器确认消息写入成功的方式。可以设置的值包括 "all"（需要所有副本确认），"1"（仅需要 Leader 副本确认），"0"（不需要确认，默认值）。使用 "all" 可以提供最高的消息可靠性，但会降低生产者的吞吐量。
         * retries：指定生产者在消息发送失败时的重试次数。当发生可重试错误时，生产者将尝试重新发送消息。默认值为 0，表示不进行重试。
         * batch.size：指定生产者在发送到同一个分区之前等待累积的消息批次大小（字节）。默认值为 16384（16KB）。较大的批次大小可以提高吞吐量，但也会增加延迟。
         * linger.ms：指定生产者在发送消息之前等待的最长时间（毫秒）。默认值为 0，表示生产者将立即发送消息而不进行等待。较大的值可以增加消息的批次大小，从而提高吞吐量，但会增加一定的延迟。
         */
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaClientParams.getBootstrapServers());
        props.put(ProducerConfig.ACKS_CONFIG, kafkaClientParams.getAcks());
        props.put(ProducerConfig.RETRIES_CONFIG, kafkaClientParams.getRetries());
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaClientParams.getBatchSize());
        props.put(ProducerConfig.LINGER_MS_CONFIG, kafkaClientParams.getLingerMs());
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaClientParams.getBufferMemory());
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, kafkaClientParams.getCompressionType());
        props.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, kafkaClientParams.getMetadataMaxAgeMs());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 创建 Kafka 生产者客户端实例
        return new KafkaProducer<>(props);
    }
}
